package com.atguigu.bigdata.edu.realtime.function;

import com.atguigu.bigdata.edu.realtime.util.*;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bigdata.edu.realtime.util.DruidDSUtil;
import com.atguigu.bigdata.edu.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;


public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimFunction<T> {
    
    private ThreadPoolExecutor threadPool;
    private DruidDataSource dataSource;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
        dataSource = DruidDSUtil.getDataSource();
        
    }
    
    // 异步调用
    @Override
    public void asyncInvoke(T input,  // 需要异步处理的数据
                            // 把处理后的数据, 放入到这个对象中
                            ResultFuture<T> resultFuture) throws Exception {
        // 补充维度代码
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                // 线程池来替我们执行 run 方法.
                // 一个线程一个客户端
                try {
                    Jedis jedis = RedisUtil.getRedisClient();
                    DruidPooledConnection conn = dataSource.getConnection();
    
                    JSONObject dim = DimUtil.readDim(jedis, conn, table(), id(input));
                    // 这行代码执行完, 这次的维度补充完毕
                    addDim(dim, input); // 把 dim 中的值, 赋值到 input 对象中  bean.setSkuName(...)
    
                    resultFuture.complete(Collections.singletonList(input));
    
                    RedisUtil.close(jedis);
                    JdbcUtil.close(conn);
                    
                } catch (SQLException e) {
                    throw new RuntimeException(e);
                }
                
                
            }
        });
        
    }
    
   
}
